
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "MQTTClient.h"
#include <unistd.h>
#include "circlebuff.h"
#include "public.h"

//#define PAYLOAD     "wgrt-data"
#define QOS         1
#define TIMEOUT     5000L
//const char *g_mqTopicDl="jianxin/baiyun_0002/control";

static volatile MQTTClient_deliveryToken deliveredtoken;

static int g_hopeToken=1;
int g_connectErrCount=0;
static MQTT_TEMP tempBuf;
// static const char *d2s="/D2S";
// static const char *rec="/S2D";

static const char *topicAFront="DHM01/HX/";
static const char *topicBFront="DHM01/JS/";
static const char *debugTopicFront="DEBUG/DHM01/";

char *gs_siteId="1234578";

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
//    int i;
    char* payloadptr;

    payloadptr = message->payload;
	pthread_mutex_lock(&comBuff0.lock);

	AP_circleBuff_WritePacket(payloadptr++,message->payloadlen,MQTPA2DTU);
	//pthread_cond_signal(&comBuff0.newPacketFlag);
	pthread_mutex_unlock(&comBuff0.lock);

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
}
void connlost(void *context, char *cause)
{
    DebugPrint("\nConnection lost cause: %s\n", cause);
}

//----------------------------------------------------------------------
void *mqttPubThread(int argc, char* argv[])
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

    MQTTClient_deliveryToken token;
    int rc, i;
	char gs_tpA_d2s[30];
	char gs_tpA_s2d[30];
	char gs_tpB_d2s[30];
	char gs_tpB_s2d[30];
	char gs_s2dDebug[30];
	//----------------------------------------------
	gs_tpA_d2s[0]=0;
   	strcat(gs_tpA_d2s,topicAFront);
   	strcat(gs_tpA_d2s,gs_siteId);
	strcat(gs_tpA_d2s,"/D2S");

	gs_tpA_s2d[0]=0;
   	strcat(gs_tpA_s2d,topicAFront);
   	strcat(gs_tpA_s2d,gs_siteId);
   	strcat(gs_tpA_s2d,"/S2D");

	gs_tpB_d2s[0]=0;
   	strcat(gs_tpB_d2s,topicBFront);
   	strcat(gs_tpB_d2s,gs_siteId);
	strcat(gs_tpB_d2s,"/D2S");

	gs_tpB_s2d[0]=0;
   	strcat(gs_tpB_s2d,topicBFront);
   	strcat(gs_tpB_s2d,gs_siteId);
   	strcat(gs_tpB_s2d,"/S2D");  

   	g_mqClientId[0]=0;
 	strcat(g_mqClientId,topicAFront);
   	strcat(g_mqClientId,gs_siteId);

	gs_s2dDebug[0]=0;
   	strcat(gs_s2dDebug,debugTopicFront);
   	strcat(gs_s2dDebug,gs_siteId);
	strcat(gs_s2dDebug,"/D2S"); 

	conn_opts.username="jxkj007";
	conn_opts.password="@nmamtf3344";
   //----------------------------------------------

    MQTTClient_create(&client, g_mqServer, g_mqClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL);

    conn_opts.keepAliveInterval = 60;
    conn_opts.cleansession = 1;

   MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

	if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
	{
		DebugPrint("start up Failed to connect, return code %d\n", rc);
	}
	MQTTClient_subscribe(client, gs_tpA_s2d, QOS);
	MQTTClient_subscribe(client, gs_tpB_s2d, QOS);


	while(1)
	{
		while(!MQTTClient_isConnected(client)){
			if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
				{
					DebugPrint(" pub retry to connect failure, return code %d\n", rc);
				}
			else
			{
				MQTTClient_subscribe(client, gs_tpA_s2d, QOS);
				MQTTClient_subscribe(client, gs_tpB_s2d, QOS);
			}
			g_connectErrCount++;
			if(g_connectErrCount>200){
				//printf(" -----system reboot--------------\n");
				//system("reboot");
			}
			sleep(3);
		}
		g_connectErrCount=0;

		while(mqSentBuff.packetSum ==0)
		{
			usleep(50000);
		}
		//----data copy from buffer-----------------------------------------------
		pthread_mutex_lock(&mqSentBuff.lock);
			
		tempBuf.len=mqSentBuff.len[mqSentBuff.rdPkIndx];
		tempBuf.topicFlag =mqSentBuff.mqttTopicFlag[mqSentBuff.rdPkIndx];
 		for(i =0;i<tempBuf.len;i++)
		{
			tempBuf.data[i]= mqSentBuff.data[mqSentBuff.rdPkIndx][i];
		}
		if(mqSentBuff.packetSum>0)mqSentBuff.packetSum--;
		if(++mqSentBuff.rdPkIndx>MAX_MQSEND_PK)mqSentBuff.rdPkIndx=0;
		//printf(" 2-----MQ SENT COPY2---mqSentBuff.packetSum=%d-----------\n",mqSentBuff.packetSum);
		pthread_mutex_unlock(&mqSentBuff.lock);
		
        //------------------------------------------------------------------------
		
		if(tempBuf.topicFlag == DTU2MQTPA)
		{
			MQTTClient_publish(client, gs_tpA_d2s,tempBuf.len, tempBuf.data,QOS,0, &token);
			DebugPrint("gs_d2s=%s-tempBuf.len=%d--g_mqServer=%s--------\n",gs_tpA_d2s,tempBuf.len,g_mqServer);
			rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
			if(rc!=0 && token!=g_hopeToken)
			{
				//printf("retry-rc=%d--token=%d--g_hopeToken=%d\n",rc,token,g_hopeToken);
				pthread_mutex_lock(&comBuff0.lock);
				AP_circleBuff_WritePacket(tempBuf.data,tempBuf.len,DTU2MQTPA);
				pthread_mutex_unlock(&comBuff0.lock);
			}
			g_hopeToken=token+1;
		}
		else if(tempBuf.topicFlag == DTU2MQTPB)
		{
			MQTTClient_publish(client, gs_tpB_d2s,tempBuf.len, tempBuf.data,QOS,0, &token);
			//printf("gs_d2s=%s-tempBuf.len=%d--g_mqServer=%s--------\n",gs_d2s,tempBuf.len,g_mqServer);
			rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
			if(rc!=0 && token!=g_hopeToken)
			{
				//printf("retry-rc=%d--token=%d--g_hopeToken=%d\n",rc,token,g_hopeToken);
				pthread_mutex_lock(&comBuff0.lock);
				AP_circleBuff_WritePacket(tempBuf.data,tempBuf.len,DTU2MQTPB);
				pthread_mutex_unlock(&comBuff0.lock);
			}
			g_hopeToken=token+1;
		}
		else if(tempBuf.topicFlag == DTU2MQTPC)
		{
			//printf("gs_s2dDebug=%s-tempBuf.len=%d--tempBuf.data=%s--------\n",gs_s2dDebug,tempBuf.len,tempBuf.data);
			MQTTClient_publish(client, gs_s2dDebug,tempBuf.len, tempBuf.data,QOS,0, &token);
			rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
			g_hopeToken=token+1;
		}
		else
		{
			DebugPrint("err publish topic topic=%d\n",tempBuf.topicFlag);
		}
	}
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
}


